package it.avantsoft.ls.peer;

import org.apache.log4j.Logger;
import it.avantsoft.ls.peer.command.PeerCommand;

import java.io.DataInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Awaits GroupLeader's messages, generates commands and enqueue them in the shared object commandQueue.
 * It uses InputStream and notifies PeerHandler when a new command has been added to the queue.
 */
public class ReceiverCommandFromGL extends Thread {
    private static Logger log = Logger.getLogger(ReceiverCommandFromGL.class);

    private boolean stop;

    private DataInputStream dataInputStream;
    private LinkedBlockingQueue<PeerCommand> commandQueue;

    private PeerHandler peerHandler;

    /**
     * ReceiverCommandFromGL basic constructor
     *
     * @param input        The InputStream of the tcp socket
     * @param commandQueue The shared CommandQueue
     * @param peerHandler  Peer Handler
     */
    public ReceiverCommandFromGL(PeerHandler peerHandler, InputStream input, LinkedBlockingQueue<PeerCommand> commandQueue) {
        super("ReceiverCommandFromGL");
        this.peerHandler = peerHandler;

        this.dataInputStream = new DataInputStream(input);
        this.commandQueue = commandQueue;
        stop = false;
    }

    /**
     * Reads input from inputstream generates a command and enqueues it in the command queue
     */
    public void run() {
        while (!stop) {
            enqueue(readCommandFromInputStream());
        }
    }

    /**
     * Waits for the commandQueue's lock then adds the new generated command and notifies
     *
     * @param peerCommand The new command to be added
     */
    private void enqueue(PeerCommand peerCommand) {
        if (peerCommand != null) {
            try {
                commandQueue.put(peerCommand);
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }
    }

    /**
     * Reads and parse messages from input stream. It creates an ObjectOutputStream on the InputStream and reads objectsw
     *
     * @return A PeerCommand
     */
    private PeerCommand readCommandFromInputStream() {
        PeerCommand command = null;
        try {
            byte commandByte = dataInputStream.readByte();
            command = PeerCommandFactory.getCommand(peerHandler, commandByte, dataInputStream);
        } catch (EOFException e) {
            stop = true;

            checkDeath();
        } catch (IOException e) {
            stop = true;
            //log.info(peerHandler.getNumber() + " " + peerHandler.getPeerId() + " stopped", e);
        }

        return command;
    }

    private void checkDeath() {
        if (peerHandler.getPeerState().equals(PeerHandler.PeerState.CONNECTED)) {
            peerHandler.deadGL();
            log.info("Changing State : " + peerHandler.getPeerState());
        } else {
            log.info(peerHandler.getPeerId() + " - Killed -> PeerState = " + peerHandler.getPeerState());
        }
    }
}
